package handler

import (
	"encoding/json"
	"fmt"
	"github.com/Shopify/sarama"
	"go.uber.org/zap"
	"io/ioutil"
	"kee/dao"
	"kee/model"
	"kee/utils/zaplog"
	"net/http"
	"regexp"
	"time"

	"github.com/tidwall/gjson"
)

// 请求eastMoney
func (impl *ImplKee) RequestEastMoneyData() error {
	unix := time.Now().Unix()
	cb := fmt.Sprintf("jQuery1124037885177771931944_%d", unix)
	pn := 1
	var rsp *http.Response
	for {
		url := fmt.Sprintf("http://21.push2.eastmoney.com/api/qt/clist/get?cb=%s&pn=%d&pz=20&po=1&np=1&ut=bd1d9ddb04089700cf9c27f6f7426281&fltt=2&invt=2&fid=f3&fs=m:0+t:6,m:0+t:80,m:1+t:2,m:1+t:23&fields=f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f12,f13,f14,f15,f16,f17,f18,f20,f21,f23,f24,f25,f22,f11,f62,f128,f136,f115,f152&_=%d", cb, pn, unix+1000)
		rsp, err := http.Get(url)
		if err != nil {
			return fmt.Errorf("RequestEastMoney|http.Get|err=%v", err)
		}
		body, err := ioutil.ReadAll(rsp.Body)
		if err != nil {
			return fmt.Errorf("RequestEastMoney|ioutil.ReadAll|err=%v", err)
		}
		err = ParseRspData(impl.Dao.ProducerServerKafka, cb, string(body), impl.Config.KafkaClientConfig.Topic)
		if err != nil {
			return fmt.Errorf("RequestEastMoney|ParseRspData|err=%v", err)
		}
		pn++
		time.Sleep(2 * time.Second)
	}

	defer rsp.Body.Close()
	return nil
}

// 解析响应的数据
func ParseRspData(syncProducerKafka *dao.SyncProducerKafka, cb, body, topic string) error {

	re := regexp.MustCompile(fmt.Sprintf(`%s\((.*)\)`, cb))
	result := re.FindAllStringSubmatch(string(body), -1)

	for _, item := range result {
		diffList := gjson.Get(item[1], "data.diff").Array()

		if len(diffList) <= 0 {
			fmt.Println("crawl end...")
			return fmt.Errorf("crawl end")
		}

		for _, value := range diffList {
			// 最新价
			newPrice := value.Get("f2").Float()
			// 涨跌幅
			priceLimit := value.Get("f3").Float()
			// 涨跌额
			upAndDown := value.Get("f4").Float()
			// 成交量
			volume := value.Get("f5").Int()
			// 成交额
			turnVolume := value.Get("f6").Float()
			// 振幅
			swing := value.Get("f7").Float()
			// 代码
			code := value.Get("f12").String()
			// 名称
			name := value.Get("f14").String()
			// 最高
			highest := value.Get("f15").Float()
			// 最低
			lowest := value.Get("f16").Float()
			// 今开
			open := value.Get("f17").Float()
			// 昨收
			closedYesterday := value.Get("f18").Float()
			// 量比
			quantityRatio := value.Get("f10").Float()
			// 换手率
			turnoverRate := value.Get("f8").Float()
			// 市盈率(动态)
			PERatioDynamic := value.Get("f9").Float()
			// 市净率
			PBRatio := value.Get("f23").Float()
			t := time.Now()
			stock := model.Stock{
				NewPrice:        newPrice,
				PriceLimit:      priceLimit,
				UpAndDown:       upAndDown,
				Volume:          int32(volume),
				TurnVolume:      turnVolume,
				Swing:           swing,
				Code:            code,
				Name:            name,
				Highest:         highest,
				Lowest:          lowest,
				Open:            open,
				ClosedYesterday: closedYesterday,
				QuantityRatio:   quantityRatio,
				TurnoverRate:    turnoverRate,
				PERatioDynamic:  PERatioDynamic,
				PBRatio:         PBRatio,
				GetDate:         t.Format("2006-01-02 15:04:05"),
			}
			stockBytes, err := json.Marshal(stock)
			if err != nil {
				zaplog.Trace("RequestEastMoneyData").Error("ParseRspData|json.Marshal", zap.Error(err))
				continue
			}

			// kafka
			partition, offset, err := syncProducerKafka.SyncProducerKafka.SendMessage(&sarama.ProducerMessage{
				Topic: topic,
				Key:   nil,
				Value: sarama.StringEncoder(string(stockBytes)),
			})

			if err != nil {
				zaplog.Trace("RequestEastMoneyData").
					Error("ParseRspData|SendMessage",
						zap.Error(err),
						zap.Int32("partition", partition),
						zap.Int64("offset", offset),
					)
				continue
			}

		}

	}
	return nil
}
